package com.rrd.dw.mr.gzip.kafka;

import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.tianxia.*;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


/*
 * 天下信用报告记录-司法
 *
 * @author:szw
 * */

public class TxCreditSiFaParse extends ParseBase {
    private Text key = new Text();
    private Text val = new Text();

    /**
     * 天下信用报告记录-司法
     *
     * @param context
     * @param model
     * @param successful
     * @return boolean
     * @throws IOException
     * @throws InterruptedException
     * @author typole
     * @date 2020/12/01 5:08 下午
     */

    @Override
    public boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        TxCreditReportRecord obj = (TxCreditReportRecord) model;
        TxCreditReportData data = obj.getData();
        StringBuilder sb = getBaseInfo(obj, successful);
        StringBuilder TxSb = new StringBuilder(500);

        RiskCaseInfo riskCaseInfo = data.getRiskCaseInfo();
        if (riskCaseInfo != null) {
            List<RiskCaseInfoData> riskCaseInfoDataList = riskCaseInfo.getData() == null ? new ArrayList<>() : riskCaseInfo.getData();
            if (!riskCaseInfoDataList.isEmpty()) {
                for (RiskCaseInfoData rc : riskCaseInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rc.getTitle() == null ? "" : rc.getTitle()).append(CRT001);
                    TxSb.append(rc.getClientType() == null ? "" : rc.getClientType()).append(CRT001);
                    TxSb.append(rc.getDecisionDate() == null ? "" : rc.getDecisionDate()).append(CRT001);
                    TxSb.append("riskCaseInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskJusticeInfo riskJusticeInfo = data.getRiskJusticeInfo();
        if (riskJusticeInfo != null) {
            List<RiskJusticeInfoData> riskJusticeInfoDataList = riskJusticeInfo.getData() == null ? new ArrayList<>() : riskJusticeInfo.getData();
            if (!riskJusticeInfoDataList.isEmpty()) {
                for (RiskJusticeInfoData rj : riskJusticeInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rj.getTitle() == null ? "" : rj.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rj.getFilingTime() == null ? "" : rj.getFilingTime()).append(CRT001);
                    TxSb.append("riskJusticeInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskPromiseInfo riskPromiseInfo = data.getRiskPromiseInfo();
        if (riskPromiseInfo != null) {
            List<RiskPromiseInfoData> riskPromiseInfoDataList = riskPromiseInfo.getData() == null ? new ArrayList<>() : riskPromiseInfo.getData();
            if (!riskPromiseInfoDataList.isEmpty()) {
                for (RiskPromiseInfoData rp : riskPromiseInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rp.getTitle() == null ? "" : rp.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rp.getPublishTime() == null ? "" : rp.getPublishTime()).append(CRT001);
                    TxSb.append("riskPromiseInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskTaxInfo riskTaxInfo = data.getRiskTaxInfo();
        if (riskTaxInfo != null) {
            List<RiskTaxInfoData> riskTaxInfoDataList = riskTaxInfo.getData() == null ? new ArrayList<>() : riskTaxInfo.getData();
            if (!riskTaxInfoDataList.isEmpty()) {
                for (RiskTaxInfoData rt : riskTaxInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rt.getTitle() == null ? "" : rt.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rt.getPublishTime() == null ? "" : rt.getPublishTime()).append(CRT001);
                    TxSb.append("riskTaxInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskDebtInfo riskDebtInfo = data.getRiskDebtInfo();
        if (riskDebtInfo != null) {
            List<RiskDebtInfoData> riskDebtInfoDataList = riskDebtInfo.getData() == null ? new ArrayList<>() : riskDebtInfo.getData();
            if (!riskDebtInfoDataList.isEmpty()) {
                for (RiskDebtInfoData rd : riskDebtInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rd.getTitle() == null ? "" : rd.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rd.getPublishTime() == null ? "" : rd.getPublishTime()).append(CRT001);
                    TxSb.append("riskDebtInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskOverdueInfo riskOverdueInfo = data.getRiskOverdueInfo();
        if (riskOverdueInfo != null) {
            List<RiskOverdueInfoData> riskOverdueInfoDataList = riskOverdueInfo.getData() == null ? new ArrayList<>() : riskOverdueInfo.getData();
            if (!riskOverdueInfoDataList.isEmpty()) {
                for (RiskOverdueInfoData ro : riskOverdueInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(ro.getTitle() == null ? "" : ro.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(ro.getPublishTime() == null ? "" : ro.getPublishTime()).append(CRT001);
                    TxSb.append("riskDebtInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskScjgInfo riskScjgInfo = data.getRiskScjgInfo();
        if (riskScjgInfo != null) {
            List<RiskScjgInfoData> riskScjgInfoDataList = riskScjgInfo.getData() == null ? new ArrayList<>() : riskScjgInfo.getData();
            if (!riskScjgInfoDataList.isEmpty()) {
                for (RiskScjgInfoData rs : riskScjgInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rs.getTitle() == null ? "" : rs.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rs.getDate() == null ? "" : rs.getDate()).append(CRT001);
                    TxSb.append("riskScjgInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskHgxxInfo riskHgxxInfo = data.getRiskHgxxInfo();
        if (riskHgxxInfo != null) {
            List<RiskHgxxInfoData> riskHgxxInfoDataList = riskHgxxInfo.getData() == null ? new ArrayList<>() : riskHgxxInfo.getData();
            if (!riskHgxxInfoDataList.isEmpty()) {
                for (RiskHgxxInfoData rh : riskHgxxInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rh.getTitle() == null ? "" : rh.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rh.getDate() == null ? "" : rh.getDate()).append(CRT001);
                    TxSb.append("riskHgxxInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskHbxxInfo riskHbxxInfo = data.getRiskHbxxInfo();
        if (riskHbxxInfo != null) {
            List<RiskHbxxInfoData> riskHbxxInfoDataList = riskHbxxInfo.getData() == null ? new ArrayList<>() : riskHbxxInfo.getData();
            if (!riskHbxxInfoDataList.isEmpty()) {
                for (RiskHbxxInfoData rhb : riskHbxxInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rhb.getTitle() == null ? "" : rhb.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rhb.getDate() == null ? "" : rhb.getDate()).append(CRT001);
                    TxSb.append("riskHbxxInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskXzcfInfo riskXzcfInfo = data.getRiskXzcfInfo();
        if (riskXzcfInfo != null) {
            List<RiskXzcfInfoData> riskXzcfInfoDataList = riskXzcfInfo.getData() == null ? new ArrayList<>() : riskXzcfInfo.getData();
            if (!riskXzcfInfoDataList.isEmpty()) {
                for (RiskXzcfInfoData rx : riskXzcfInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rx.getTitle() == null ? "" : rx.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rx.getDate() == null ? "" : rx.getDate()).append(CRT001);
                    TxSb.append("riskXzcfInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskZbajInfo riskZbajInfo = data.getRiskZbajInfo();
        if (riskZbajInfo != null) {
            List<RiskZbajInfoData> riskZbajInfoDataList = riskZbajInfo.getData() == null ? new ArrayList<>() : riskZbajInfo.getData();
            if (!riskZbajInfoDataList.isEmpty()) {
                for (RiskZbajInfoData rz : riskZbajInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rz.getTitle() == null ? "" : rz.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rz.getDate() == null ? "" : rz.getDate()).append(CRT001);
                    TxSb.append("riskZbajInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskGgsdInfo riskGgsdInfo = data.getRiskGgsdInfo();
        if (riskGgsdInfo != null) {
            List<RiskGgsdInfoData> riskGgsdInfoDataList = riskGgsdInfo.getData() == null ? new ArrayList<>() : riskGgsdInfo.getData();
            if (!riskGgsdInfoDataList.isEmpty()) {
                for (RiskGgsdInfoData rg : riskGgsdInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rg.getTitle() == null ? "" : rg.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rg.getDate() == null ? "" : rg.getDate()).append(CRT001);
                    TxSb.append("riskGgsdInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskKtggInfo riskKtggInfo = data.getRiskKtggInfo();
        if (riskKtggInfo != null) {
            List<RiskKtggInfoData> riskKtggInfoDataList = riskKtggInfo.getData() == null ? new ArrayList<>() : riskKtggInfo.getData();
            if (!riskKtggInfoDataList.isEmpty()) {
                for (RiskKtggInfoData rk : riskKtggInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rk.getTitle() == null ? "" : rk.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rk.getDate() == null ? "" : rk.getDate()).append(CRT001);
                    TxSb.append("riskKtggInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        RiskSplcInfo riskSplcInfo = data.getRiskSplcInfo();
        if (riskSplcInfo != null) {
            List<RiskSplcInfoData> riskSplcInfoDataList = riskSplcInfo.getData() == null ? new ArrayList<>() : riskSplcInfo.getData();
            if (!riskSplcInfoDataList.isEmpty()) {
                for (RiskSplcInfoData rsp : riskSplcInfoDataList) {
                    TxSb.delete(0, TxSb.length());
                    TxSb.append(rsp.getTitle() == null ? "" : rsp.getTitle()).append(CRT001);
                    TxSb.append(CRT001);
                    TxSb.append(rsp.getDate() == null ? "" : rsp.getDate()).append(CRT001);
                    TxSb.append("riskSplcInfo").append(CRT001);
                    key.set(getKey());
                    val.set((sb.toString() + TxSb.toString()).replace("\n", " ").replace("\r", " "));
                    context.write(key, val);
                }
            }
        }

        return riskCaseInfo != null || riskDebtInfo != null || riskJusticeInfo != null || riskOverdueInfo != null || riskPromiseInfo != null || riskTaxInfo != null || riskScjgInfo != null || riskHgxxInfo != null || riskHbxxInfo != null || riskXzcfInfo != null || riskZbajInfo != null || riskGgsdInfo != null || riskKtggInfo != null || riskSplcInfo != null;
    }

    @Override
    public String getSubType() {
        return ".sifa";
    }

    @Override
    public String getTableName() {
        return "kafka_babel_tx_credit_sifa_inc";
    }

    @Override
    public String getType() {
        return RecordType.TX_CREDIT_REPORT_RECORD.toString();
    }

    @Override
    public String getComment() {
        return "天下信用-司法";
    }
}
